package com.example.kafkaspringboot;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @Description: Kafka 消费者
 * @Author: dengPing.xu
 * @Data: 2022-01-25
 */
public class KafkaConsumerTest {

    private static final String TOPIC = "topic-test";

    private static final String BOOTSTRAP_SERVER = "175.178.58.43:9092";

    private static final String GROUP_ID = "topic-test-consumer-01";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add(TOPIC);
        consumer.subscribe(topicList);
        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(item -> {
                System.out.println("分区数：" + item.partition() + "\n偏移量：" + item.offset()  + "\n键：" +item.key()  + "\n值：" +item.value());
            });
        }

    }
}
